package com.desheng.bigdata.flink.time

import java.net.URL

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
  * Flink中对于数据时间的语义提供了三种
  *     TimeCharacteristic.ProcessingTime
  *         存在很多的不确定性，因为获取的数据计算时间是基于当下机器的时间，在一个时间窗口操作过程中
  *         那么谁先来到，就计算谁，这显然会后很多数据的遗漏。
  *         但是，同时相较于其他两种时间，延迟最低，性能消耗最低。
  *     TimeCharacteristic.IngestionTime
  *         该时间是数据摄入flink datastream中所产生的数据，是processing和eventtime的妥协的结果。
  *         针对大多数算子，eventtime的所产生的时间已经足够使用，但是会有经度的乱序产生，但是一个乱序
  *         是在执行shuffle操作，比如split、union、join等操作是产生的。
  *     TimeCharacteristic.EventTime
  *         该时间就是事件在源头产生的时候被系统，或者用户分配的各个事件彼此隔离的时间，这些事件被采集到
  *         计算系统中不能保证数据产生的时间顺序，也就是说后产生的事件可能会比先产生的事件先抵达计算系统。
  *
  *         flink中基于event-time处理数据的时候，为了能够保证在一个时间间隔内(window)产生的数据都抵达，
  *         进而再执行window和排序的操作，flink中使用time-watermark机制来进行保证，延迟数据被处理。
  *
  *     水印主要是用来处理EventTime的。
  *     分配水印有两种方式：
  *         AssignerWithPeriodicWatermarks  ---->周期性的为一条记录设置一个水印时间，默认时间是200ms
  *         AssignerWithPunctuatedWatermarks ---非周期性，就位每一条记录设置一个水印的时间
  */
object _01TimeCharacteristicApp {
    def main(args: Array[String]): Unit = {

        val env = StreamExecutionEnvironment.getExecutionEnvironment
        //要基于eventtime进行处理
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

        val lines = env.fromCollection(List(
            UserClick(1, "张三", "http://bd.ujiuye.com", 1600239353208L),
            UserClick(2, "李四", "http://java.ujiuye.com", 1600239353608L),
            UserClick(3, "王五", "http://bd.ujiuye.com", 1600239353428L),
            UserClick(4, "赵六", "http://ui.ujiuye.com", 1600239353701L),
            UserClick(5, "周七", "http://python.ujiuye.com", 1600239352258L)
        ))

        //有界无序周期性水印
        lines.assignTimestampsAndWatermarks(
            new BoundedOutOfOrdernessTimestampExtractor[UserClick](Time.seconds(2)) {
                override def extractTimestamp(uc: UserClick): Long = {
                    uc.timestamp
                }
            }
        ).map(uc => {
            val url = new URL(uc.url)
            url.getHost
            (url.getHost, uc)
        }).keyBy(kv => kv._1)
            .timeWindow(Time.seconds(4))
            .process(new ProcessWindowFunction[(String, UserClick), (String, Int), String, TimeWindow] {
                override def process(key: String, context: Context,
                                     elements: Iterable[(String, UserClick)],
                                     out: Collector[(String, Int)]): Unit = {
                    elements.foreach{case (host, ucs) => {
                        println(ucs)
                    }}
                    out.collect((key, elements.size))
                }
            })
            .print()

        env.execute()
    }
}
case class UserClick(id: Int, name: String, url: String, timestamp: Long)